package time.processingTime;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.descriptors.*;
import org.apache.flink.types.Row;

/**
 * 定义处理时间(Processing Time)
 * 定义 Table Schema 时指定
 */
public class ProcessingTime2 {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        String topicFrom = "myTest";
        String kafkaServer = "192.168.36.130:9092,192.168.36.129:9092,192.168.36.128:9092";

        Kafka kafka = new Kafka().version("universal").topic(topicFrom).property("bootstrap.servers", kafkaServer);
        Schema schema = new Schema()
                .field("id", DataTypes.STRING())
                .field("ts", DataTypes.BIGINT())
                .field("temp", DataTypes.DOUBLE())
                .field("pt", DataTypes.TIMESTAMP(3)).proctime();
        tableEnv.connect(kafka)
                .withFormat(new Json())
                .withSchema(schema)
                .inAppendMode()
                .createTemporaryTable("inputTable");
        // 转换
        Table inputTable = tableEnv.from("inputTable");
        // 输出结构
        inputTable.printSchema();
        // 打印输出
        tableEnv.toAppendStream(inputTable, Row.class).print();

        env.execute();
    }
}
